package big.kafka.producter.asyncsend;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 异步发送消息
 */
public class AsyncKafkaApi {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop129:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        properties.put(ProducerConfig.ACKS_CONFIG,"-1"); //ack
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384); //收集器 最大缓存
        properties.put(ProducerConfig.LINGER_MS_CONFIG,1);   //send 等待毫秒数
        //1.创建一个生产者对象
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String,String>(properties);
        //2.调用send方法
        for (int i = 0; i < 1000; i++) {
            kafkaProducer.send(new ProducerRecord<String, String>("first", i + ",", "message: " + i), new Callback() {
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null ){
                        System.out.println(":sucess");
                    }else{
                        System.out.println(e.getMessage());
                    }
                }
            });
        }
        //关闭生产者
        kafkaProducer.close();
    }
}
